package drds.data_migrate.task_group.task.run;

import drds.common.$;
import drds.data_migrate.task_group.task.data_exchange.exchanger.Consume;
import drds.data_migrate.task_group.task.data_exchange.record.Record;
import drds.data_migrate.task_group.task.data_exchange.record.column.Column;
import drds.data_migrate.task_group.task.data_exchange.record.column.ColumnType;
import drds.data_migrate.util.Jdbc;
import drds.data_migrate.util.Jdbcs;
import drds.data_migrate.util.Updates;
import drds.data_migrate.vo.TaskConfig;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;

/**
 * Created by ibm on 2018/11/3.
 */
@Slf4j
public class WriteTask
{

    private static final String VALUE_HOLDER = "?";
    protected static String insert_or_replace_template;
    String writeRecordSql;
    private TaskConfig taskConfig;

    public WriteTask()
    {

    }

    public WriteTask(TaskConfig taskConfig)
    {
        this.taskConfig = taskConfig;
    }

    public void init()
    {


        /*this.tableName = taskConfig.getTableName();
        this.columnNameList = taskConfig.getColumnNameList();
        this.columnNumber = taskConfig.getColumnNumber();
        //
        this.preSqlList = taskConfig.getPreSqlList();
        this.postSqlList = taskConfig.getPostSqlList();

        this.batchSize = taskConfig.getBatchSize();
        this.batchByteSize = taskConfig.getBatchByteSize();
        //
        writeMode = taskConfig.getWriteMode();//insert
        emptyAsNull = taskConfig.isEmptyAsNull();*/

        insert_or_replace_template = taskConfig.getInsert_or_replace_template();
        this.writeRecordSql = String.format(insert_or_replace_template, this.taskConfig.getTableName());


    }

    public void prepare() throws SQLException
    {

        Connection connection = Jdbc.getConnection(taskConfig.getJdbcUrl(), taskConfig.getUsername(), taskConfig.getPassword());

        Jdbcs.sessionSet(connection, taskConfig.getSessionSqlList());

        int tableNumber = 1;
        if (tableNumber != 1)
        {

            Updates.executeSqlList(connection, this.taskConfig.getPreSqlList());
        }

        Jdbc.close(null, null, connection);
    }

    public void startWrite(Consume consume, Connection connection)
    {


        // 写数据库的SQL语句
        calcWriteRecordSql();

        List<Record> recordList = new ArrayList<Record>(this.taskConfig.getBatchSize());

        try
        {
            Record record;
            while ((record = consume.getFromRecordList()) != null)
            {
                if (record.getColumnListSize() != this.taskConfig.getColumnNumber())
                {
                    // 源头读取字段列数与目的表字段写入列数不相等，直接报错
                   /* throw DataExchangeException
                            .asDataExchangeException(
                                    DBUtilErrorCode.CONF_ERROR,
                                    String.format(
                                            "列配置信息有错误. 因为您配置的任务中，. 请检查您的配置并作出修改.",
                                            record.getColumnListSize(),
                                            this.columnNumber));*/
                    throw new RuntimeException("源头读取字段数:%s 与 目的表要写入的字段数:%s 不相等");
                }

                recordList.add(record);


                if (recordList.size() >= this.taskConfig.getBatchSize())
                {
                    batchInsert(connection, recordList);
                    recordList.clear();

                }
            }
            if (!recordList.isEmpty())
            {
                batchInsert(connection, recordList);
                recordList.clear();

            }
        } catch (Exception e)
        {
            throw new RuntimeException();
        } finally
        {
            recordList.clear();

            Jdbc.close(null, null, connection);
        }
    }

    // TODO 改用连接池，确保每次获取的连接都是可用的（注意：连接可能需要每次都初始化其 session）
    public void startWrite(Consume consume) throws SQLException
    {


        Connection connection = Jdbc.getConnection(taskConfig.getJdbcUrl(), taskConfig.getUsername(), taskConfig.getPassword());
        Jdbcs.sessionSet(connection, taskConfig.getSessionSqlList());
        startWrite(consume, connection);
    }


    public void post() throws SQLException
    {
        int tableNumber = 1;

        boolean hasPostSql = $.isHasData(this.taskConfig.getPostSqlList());
        if (tableNumber == 1 || !hasPostSql)
        {
            return;
        }

        Connection connection = Jdbc.getConnection(this.taskConfig.getJdbcUrl(), this.taskConfig.getUsername(), this.taskConfig.getPassword());


        Updates.executeSqlList(connection, this.taskConfig.getPostSqlList());
        Jdbc.close(null, null, connection);
    }

    public void destroy()
    {
    }

    protected void batchInsert(@NonNull Connection connection, List<Record> recordList)
            throws SQLException
    {
        PreparedStatement preparedStatement = null;
        try
        {
            connection.setAutoCommit(false);
            preparedStatement = connection.prepareStatement(this.writeRecordSql);
            for (Record record : recordList)
            {
                preparedStatement = setValue(preparedStatement, record);
                preparedStatement.addBatch();
            }
            preparedStatement.executeBatch();
            connection.commit();
        } catch (SQLException e)
        {
            log.warn("回滚此次写入, 采用每次写入一行方式提交. 因为:" + e.getMessage());
            connection.rollback();
            insert(connection, recordList);
        } catch (Exception e)
        {
            throw new RuntimeException();
        } finally
        {
            Jdbc.close(preparedStatement, null);
        }
    }

    protected void insert(Connection connection, List<Record> recordList)
    {
        PreparedStatement preparedStatement = null;
        try
        {
            connection.setAutoCommit(true);
            preparedStatement = connection.prepareStatement(this.writeRecordSql);
            for (Record record : recordList)
            {
                try
                {
                    preparedStatement = setValue(preparedStatement, record);
                    preparedStatement.execute();
                } catch (SQLException e)
                {
                    log.debug(e.toString());
                    //this.abstractTaskPluginCollector.collectDirtyRecord(record, e);
                } finally
                {
                    preparedStatement.clearParameters();
                }
            }
        } catch (SQLException e)
        {
            throw new RuntimeException();
        } finally
        {
            Jdbc.close(preparedStatement, null);
        }
    }

    protected PreparedStatement setValue(PreparedStatement preparedStatement, Record record)
            throws SQLException
    {
        for (int columnIndex = 0; columnIndex < this.taskConfig.getColumnNumber(); columnIndex++)
        {
            preparedStatement = setValue(preparedStatement, columnIndex, record.getColumn(columnIndex));
        }

        return preparedStatement;
    }

    protected PreparedStatement setValue(PreparedStatement preparedStatement, int columnIndex, Column column) throws SQLException
    {
        //在制定迁移计划前需要保证两边的表结构要一致。如果不能保证所有的字段都一致的话则不允许迁移.


        if (ColumnType.NULL == column.getColumnType())
        {
            preparedStatement.setNull(columnIndex, column.asInteger());
        } else if (ColumnType.BOOLEAN == column.getColumnType())
        {
            preparedStatement.setBoolean(columnIndex, column.asBoolean());
        } else if (ColumnType.SHORT == column.getColumnType())
        {
            preparedStatement.setShort(columnIndex, column.asShort());
        } else if (ColumnType.INT == column.getColumnType())
        {
            preparedStatement.setInt(columnIndex, column.asInteger());
        } else if (ColumnType.LONG == column.getColumnType())
        {
            preparedStatement.setLong(columnIndex, column.asLong());
        } else if (ColumnType.STRING == column.getColumnType())
        {
            preparedStatement.setString(columnIndex, column.asString());
        } else if (ColumnType.FLOAT == column.getColumnType())
        {
            preparedStatement.setFloat(columnIndex, column.asFloat());
        } else if (ColumnType.DOUBLE == column.getColumnType())
        {
            preparedStatement.setDouble(columnIndex, column.asDouble());
        } else if (ColumnType.BIG_DECIMAL == column.getColumnType())
        {
            preparedStatement.setBigDecimal(columnIndex, column.asBigDecimal());
        } else if (ColumnType.STRING == column.getColumnType())
        {
            preparedStatement.setString(columnIndex, column.asString());
        } else if (ColumnType.DATE == column.getColumnType())
        {
            preparedStatement.setDate(columnIndex, column.asDate());
        } else if (ColumnType.TIME == column.getColumnType())
        {
            preparedStatement.setTime(columnIndex, column.asTime());
        } else if (ColumnType.TIMESTAMP == column.getColumnType())
        {
            preparedStatement.setTimestamp(columnIndex, column.asTimestamp());
        } else
        {
            throw new IllegalArgumentException("该方法不支持!");
        }

        return preparedStatement;
    }

    private void calcWriteRecordSql()
    {
        boolean forceUseUpdate = false;
        List<String> valueHolders = new ArrayList<String>();
        insert_or_replace_template = Updates.getWriteTemplate(taskConfig.getColumnNameList(), valueHolders, this.taskConfig.getWriteMode(), forceUseUpdate);
        writeRecordSql = String.format(insert_or_replace_template, this.taskConfig.getTableName());
    }


}
